package com.atguigu.gmall.realtime.utils;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;
import java.util.Properties;

/**
 * TODO
 *
 * @author cjp
 * @version 1.0
 * @date 2021/9/13 11:13
 */
public class MyKafkaUtil {

    private static final String KAFKA_SERVERS = "hadoop202:9092,hadoop203:9092,hadoop204:9092";
    private static final String DEFAULT_TOPIC = "default_topic";

    /**
     * 获取 读取 kafka的 KafkaSourceFunction
     *
     * @param topic
     * @param groupId
     * @return
     */
    public static FlinkKafkaConsumer<String> getKafkaSourceFunction(String topic, String groupId) {

        Properties prop = new Properties();
        prop.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_SERVERS);
        prop.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);

        return new FlinkKafkaConsumer<>(topic, new SimpleStringSchema(), prop);

    }

    /*public static FlinkKafkaProducer<String> getKafkaSinkFunction(String topic) {

        Properties producerProp = new Properties();
        producerProp.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_SERVERS);
//        producerProp.setProperty(ProducerConfig.ACKS_CONFIG, "-1");
        return new FlinkKafkaProducer<String>(topic, new SimpleStringSchema(), producerProp);
    }*/

    public static FlinkKafkaProducer<String> getKafkaSinkFunction(String topic) {

        Properties producerProp = new Properties();
        producerProp.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_SERVERS);
        producerProp.setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 1000 * 60 * 15 + "");
//        producerProp.setProperty(ProducerConfig.ACKS_CONFIG, "-1");
        return new FlinkKafkaProducer<String>(DEFAULT_TOPIC, new KafkaSerializationSchema<String>() {
            @Override
            public ProducerRecord<byte[], byte[]> serialize(String str, @Nullable Long timestamp) {
                return new ProducerRecord<byte[], byte[]>(topic,str.getBytes());
            }
        }, producerProp, FlinkKafkaProducer.Semantic.EXACTLY_ONCE);
    }

    public static <T> FlinkKafkaProducer<T> getKafkaSinkFunctionBySchema(KafkaSerializationSchema<T> kafkaSerializationSchema) {

        Properties producerProp = new Properties();
        producerProp.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_SERVERS);
        producerProp.setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 1000 * 60 * 15 + "");
        //producerProp.setProperty(ProducerConfig.ACKS_CONFIG, "-1");

        return new FlinkKafkaProducer<T>(DEFAULT_TOPIC, kafkaSerializationSchema, producerProp, FlinkKafkaProducer.Semantic.EXACTLY_ONCE);
    }

    public static String getKafkaDDL(String topic, String groupId) {
        String ddl = "  'connector' = 'kafka'," +
            "  'topic' = '" + topic + "'," +
            "  'properties.bootstrap.servers' = '" + KAFKA_SERVERS + "'," +
            "  'properties.group.id' = '" + groupId + "'," +
            "  'scan.startup.mode' = 'latest-offset'," +
            "  'format' = 'json'";
        return ddl;
    }
}
